package org.apache.beam.examples.jdbc;

import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.jdbc.JdbcIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.values.KV;

import java.sql.ResultSet;

public class BeamH2Example {

    static final Log log = LogFactory.get();

    public static void main(String[] args) {

        PipelineOptions options = PipelineOptionsFactory.fromArgs(args).withValidation().create();
        Pipeline pipeline = Pipeline.create(options);

        String query = "SELECT * FROM sys_user";
        pipeline.apply(JdbcIO.<KV<String, String>>read()
                .withDataSourceConfiguration(JdbcIO.DataSourceConfiguration.create(
                                H2DataBaseInit.driver, H2DataBaseInit.url)
                        .withUsername(H2DataBaseInit.user)
                        .withPassword(H2DataBaseInit.password))
                .withQuery(query)
                .withRowMapper(new JdbcIO.RowMapper<KV<String, String>>() {
                    public KV<String, String> mapRow(ResultSet resultSet) throws Exception {
                        log.info("获取到数据了:{}", resultSet.getString(1));
                        return KV.of(resultSet.getString(1), resultSet.getString(2));
                    }
                })
        )/*.apply(JdbcIO.<KV<String, String>>write()
                .withDataSourceConfiguration(JdbcIO.DataSourceConfiguration.create(
                                "org.h2.Driver", "jdbc:h2:F:/DATA/demo/demo")
                        .withUsername("SA")
                        .withPassword(""))
                .withStatement("insert into SYS_USER (ID, ACCOUNT, CNNAME, EMAIL, MOBILE, USERSTATE, ORGCODE, PWD, CREATETIME, CERTTYPE, CERTNO)\n" +
                        "values (?,?,?,?,?,?,?,?,?,?,?)")
                .withPreparedStatementSetter(new JdbcIO.PreparedStatementSetter<KV<String, String>>() {
                    public void setParameters(KV<String, String> element, PreparedStatement query)
                            throws SQLException {
                        query.setString(1, element.getKey());
                        query.setString(2, element.getValue());
                        query.setString(3, element.getValue());
                        query.setString(4, element.getValue());
                        query.setString(5, element.getValue());
                        query.setString(6, element.getValue());
                        query.setString(7, element.getValue());
                        query.setString(8, element.getValue());
                        query.setString(9, element.getValue());
                        query.setString(10, element.getValue());
                        query.setString(11, element.getValue());
                    }
                })
        )*/;
        pipeline.run().waitUntilFinish();
    }
}